package com.shujia.flink.core;

import com.shujia.flink.bean.Alert;
import com.shujia.flink.bean.Transaction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

public class Demo9Transaction {
    public static void main(String[] args) throws Exception {
        /*
         * 001,100
         * 001,200
         * 001,0.01
         * 001,600
         * 001,200
         * 001,0.2
         * 001,300
         * 001,700
         */

        /*
         * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Transaction> transDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String id = split[0];
            double price = Double.parseDouble(split[1]);
            return new Transaction(id, price);
        });

        //按照账户id分组
        KeyedStream<Transaction, String> keyByDS = transDS.keyBy(Transaction::getId);

        //每隔一条数据计算最近两条数据
        WindowedStream<Transaction, String, GlobalWindow> windowDS = keyByDS
                .countWindow(2, 1);

        SingleOutputStreamOperator<Alert> alertsDS = windowDS
                .apply(new WindowFunction<Transaction, Alert, String, GlobalWindow>() {
                    @Override
                    public void apply(String id, //账户编号
                                      GlobalWindow window,//窗口
                                      Iterable<Transaction> transactions, //窗口内所有的数据
                                      Collector<Alert> out//用于将数据发送到下游
                    ) throws Exception {
                        //对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
                        ArrayList<Transaction> transactionArrayList = new ArrayList<>();
                        for (Transaction transaction : transactions) {
                            transactionArrayList.add(transaction);
                        }

                        if (transactionArrayList.size() == 2) {
                            //获取第一条和第二条
                            Transaction head = transactionArrayList.get(0);
                            Transaction last = transactionArrayList.get(1);

                            if (head.getPrice() < 1 && last.getPrice() > 500) {
                                //发出告警信息
                                Alert alert = new Alert();
                                alert.setId(id);
                                alert.setHeadPrice(head.getPrice());
                                alert.setLasePrice(last.getPrice());

                                //发送到下游
                                out.collect(alert);
                            }
                        }
                    }
                });

        alertsDS.print();

        env.execute();

    }
}


